package com.bj58.mis.apm.collector.kydata;

import com.bj58.mis.apm.client.core.AppNodeContext;
import com.bj58.mis.apm.client.core.util.JsonUtil;
import com.bj58.mis.apm.collector.BaseSchedulableCollector;
import org.apache.http.client.methods.CloseableHttpResponse;
import org.apache.http.client.methods.HttpGet;
import org.apache.http.client.methods.HttpPost;
import org.apache.http.client.methods.HttpUriRequest;
import org.apache.http.config.Registry;
import org.apache.http.config.RegistryBuilder;
import org.apache.http.conn.socket.ConnectionSocketFactory;
import org.apache.http.conn.socket.LayeredConnectionSocketFactory;
import org.apache.http.conn.socket.PlainConnectionSocketFactory;
import org.apache.http.conn.ssl.SSLConnectionSocketFactory;
import org.apache.http.impl.client.CloseableHttpClient;
import org.apache.http.impl.client.HttpClients;
import org.apache.http.impl.conn.PoolingHttpClientConnectionManager;
import org.apache.http.message.BasicHeader;
import org.apache.http.util.EntityUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.net.ssl.SSLContext;
import java.io.IOException;
import java.security.NoSuchAlgorithmException;
import java.util.*;

/**
 * kylin数据采集器
 * @author : wangxw
 */
public class KylinDataCollector extends BaseSchedulableCollector<KylinData> {

    private Logger logger = LoggerFactory.getLogger(this.getClass());

    private final String baseURL = AppNodeContext.getValue("kylin.server.url");
    private final String kylinUsername = AppNodeContext.getValue("kylin.server.username");
    private final String kylinPassword = AppNodeContext.getValue("kylin.server.password");
    private final String serverURL = AppNodeContext.getValue("server.http.url");

    private PoolingHttpClientConnectionManager connectionManager = null;

    public String listQueryableTables(String cookie, String projectName){
        String method = "GET";
        String para = "/tables_and_columns?project="+projectName;
        return excuteGet(cookie, para);
    }
    public String listProject(String cookie, Integer offset, Integer limit){
        String method = "GET";
        StringBuilder para = new StringBuilder();
        para.append("/projects?");
        if(offset != null && offset >= 0){
            para.append("&offset="+offset);
        }
        if(limit != null && limit > 0){
            para.append("&limit="+limit);
        }
        return excuteGet(cookie, para.toString());
    }
    public String listCubes(String cookie, Integer offset,Integer limit,String cubeName, String projectName ){
        String method = "GET";
        StringBuilder para = new StringBuilder();
        para.append("/cubes?");
        if(offset != null && offset >= 0){
            para.append("&offset="+offset);
        }
        if(limit != null && limit > 0){
            para.append("&limit="+limit);
        }
        if(cubeName != null && !cubeName.equals("")){
            para.append("&cubeName="+cubeName);
        }
        if(projectName != null && !projectName.equals("")){
            para.append("&projectName="+projectName);
        }
        return excuteGet(cookie, para.toString());
    }

    public String getCubeDesc(String cookie, String cubeName){
        String method = "GET";
        String para = "/cube_desc/"+cubeName+"/desc";
        return excuteGet(cookie, para);
    }

    public String getCube(String cookie, String cubeName){
        String method = "GET";
        String para = "/cubes/"+cubeName;
        return excuteGet(cookie, para);
    }

    public String getDataModel(String cookie, String modelName){
        String method = "GET";
        String para = "/model/"+modelName;
        return excuteGet(cookie, para);
     }

    public String getJobs(String cookie, String jobId, String cubeName){
        String method = "GET";
        String para = "/jobs/"+jobId;
        if(cubeName != null){
            para = "/jobs?cubeName="+cubeName;
        }
        return excuteGet(cookie, para);
     }

    public String getJobsOutput(String cookie, String jobId, String setpId){
        String method = "GET";
        String para = "/jobs/"+jobId+"/steps/"+setpId+"/output";
        return excuteGet(cookie, para);
    }

     //http://10.126.89.86:7070/kylin/api/jobs/c1360c3f-9975-48c0-b2d9-7250deae8023/steps/c1360c3f-9975-48c0-b2d9-7250deae8023-00/output

    public String getHiveTables(String cookie, String projectName){
        String method = "GET";
//        String para = "/tables?project="+projectName+"&ext=true";
        String para = "/tables?project="+projectName;
        return excuteGet(cookie, para);
    }

    public String getCuboids(String cookie, String cubeName){
        String para = "/cubes/"+cubeName+"/cuboids/current";
        return excuteGet(cookie, para);
    }

    private String excuteGet(String cookie, String para){
        HttpGet httpGet = new HttpGet(baseURL+para);
        httpGet.addHeader(new BasicHeader("Content-Type", "application/json"));
        httpGet.addHeader(new BasicHeader("Cookie", cookie));
        return execute(httpGet);
    }

    private String execute(HttpUriRequest request) {
        CloseableHttpClient httpClient = getHttpClient();
        String json = null;
        CloseableHttpResponse response=null;
        try {
            response = httpClient.execute(request);
            json = EntityUtils.toString(response.getEntity(), "UTF-8");
        } catch (UnsupportedOperationException e) {
            e.printStackTrace();
        } catch (IOException e) {
            e.printStackTrace();
        }finally {
            if(response!=null){
                try {
                    response.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }
        return json;
    }

    /**
     * 初始化连接池并返回连接
     * @return
     */
    private CloseableHttpClient getHttpClient() {
        if(connectionManager == null){
            LayeredConnectionSocketFactory sslsf = null;
            try {
                sslsf = new SSLConnectionSocketFactory(SSLContext.getDefault());
            } catch (NoSuchAlgorithmException e) {
                e.printStackTrace();
            }


            Registry<ConnectionSocketFactory> socketFactoryRegistry = RegistryBuilder.<ConnectionSocketFactory> create()
                    .register("https", sslsf)
                    .register("http", new PlainConnectionSocketFactory())
                    .build();
            connectionManager =new PoolingHttpClientConnectionManager(socketFactoryRegistry);
            connectionManager.setMaxTotal(200);
            connectionManager.setDefaultMaxPerRoute(20);
        }
        return HttpClients.custom().setConnectionManager(connectionManager).build();
    }

    public String getServerUpTime() throws Exception{
        HttpGet httpGet = new HttpGet(serverURL + "/kylin/lasttime");
        httpGet.addHeader(new BasicHeader("Content-Type", "application/json"));
        return execute(httpGet);
    }

    public String getLoginCookie(String user,String passwd){
        HttpPost httpPost = new HttpPost(baseURL + "/user/authentication");
        byte[] key = (user+":"+passwd).getBytes();
        String encoding = new sun.misc.BASE64Encoder().encode(key);
        httpPost.addHeader(new BasicHeader("Authorization", "Basic " + encoding));
        httpPost.addHeader(new BasicHeader("Content-Type", "application/json"));


        CloseableHttpClient httpClient = getHttpClient();
        String returnCookie = null;
        CloseableHttpResponse response=null;
        try {
            response = httpClient.execute(httpPost);
            String cookie = response.getFirstHeader("Set-Cookie").getValue();
            if(cookie != null && cookie.indexOf("JSESSIONID") > -1){
                returnCookie = cookie.substring(0, cookie.indexOf(";"));
            }
        } catch (UnsupportedOperationException e) {
            e.printStackTrace();
        } catch (IOException e) {
            e.printStackTrace();
        }finally {
            if(response!=null){
                try {
                    response.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }
        return returnCookie;
    }

    @Override
    public void schedule() {
        KylinData data = queryKylinData();
        if(data != null) {
            super.collect(data);
        }
    }

    public KylinData queryKylinData(){
        String cookie = getLoginCookie(this.kylinUsername,this.kylinPassword);
        if(cookie == null){
            throw new RuntimeException("登陆kylin验证失败，登陆信息无效。");
        }
        long lasttime = 0;
        try{
            String json = getServerUpTime();
            if(json != null && json.contains("lasttime")){
                lasttime = ((Double)JsonUtil.fromJson(json).get("lasttime")).longValue();
            }
        }catch (Exception e){
            logger.error(e.getMessage());
            return null;
        }

        KylinData kylinData = new KylinData();

        //1. 处理项目
        String projectjson = listProject(cookie, null, null);
        List<Map<String, Object>> projects = JsonUtil.fromJsonArray(projectjson);
        Iterator<Map<String, Object>> it = projects.iterator();
        List<Map<String, Object>> cubeDescs = new ArrayList<Map<String, Object>>();
        List<Map<String, Object>> cubes = new ArrayList<>();
        List<Map<String, Object>> tables = new ArrayList<Map<String, Object>>();
        List<Map<String, Object>> models = new ArrayList<Map<String, Object>>();
        while(it.hasNext()){
            Map<String, Object> map = it.next();
            /*
            if(((Long)map.get("last_modified")) <= lasttime){
                it.remove();
            }
            */

            //查询项目下的hive表
            String tablejson = getHiveTables(cookie, (String)map.get("name"));
            List<Map<String, Object>> subtables = JsonUtil.fromJsonArray(tablejson);
            for(Map<String, Object> t : subtables){
                t.put("projectId", map.get("uuid"));
            }
            tables.addAll(subtables);
            //查询项目下model
            List<String> ms = (List<String>)map.get("models");
            Map<String, String> modelTemp = new HashMap<>();
            if(ms != null && ms.size() > 0){
                for(String modelName : ms){
                    String modeljson = getDataModel(cookie, modelName);
                    Map<String, Object> model = JsonUtil.fromJson(modeljson);
                    model.put("projectId", map.get("uuid"));
                    models.add(model);

                    modelTemp.put((String)model.get("name"), (String)model.get("uuid"));
                }
            }

            //根据项目关联cubedesc和cube查询
            List<Map<String, Object>> realizations = (List<Map<String, Object>>)map.get("realizations");
            if(realizations != null && realizations.size() > 0){
                for(Map<String, Object> m : realizations){
                    if("CUBE".equals(m.get("type"))){
                        String name = (String)m.get("realization");
                        if(name == null || name.equals("")){
                            continue;
                        }

                        String cdjson = getCubeDesc(cookie, name);
                        Map<String, Object> cubedesc = JsonUtil.fromJson(cdjson);
                        if(cubedesc != null) {
                            cubedesc.put("projectId", map.get("uuid"));
                            cubedesc.put("modelId", modelTemp.get(cubedesc.get("model_name")));
                            cubeDescs.add(cubedesc);
                        }

                        String cjson = getCube(cookie, name);
                        Map<String, Object> cube = JsonUtil.fromJson(cjson);
                        if(cube != null) {
                            cube.put("projectId", map.get("uuid"));
                            cube.put("cubedescId", cubedesc == null ? null : cubedesc.get("uuid"));
                            cubes.add(cube);
                        }
                    }
                }
            }

        }
        kylinData.setProjects(projects);

        //2.处理cubedesc
        it = cubeDescs.iterator();
        while(it.hasNext()){
            Map<String, Object> map = it.next();
            if(map == null || map.get("last_modified") == null || ((Double)map.get("last_modified")).longValue() <= lasttime){
                it.remove();
            }
        }
        kylinData.setCubeDescs(cubeDescs);

        //3.处理cube
        it = cubes.iterator();
        List<Map<String, Object>> jobs = new ArrayList<Map<String, Object>>();
        List<Map<String, Object>> cuboids = new ArrayList<Map<String, Object>>();
        while(it.hasNext()){
            Map<String, Object> map = it.next();
            if(map == null || map.get("last_modified") == null || ((Double)map.get("last_modified")).longValue() <= lasttime){
                it.remove();
            }else{
                /*根据cubeName获取所有执行任务数据*/
                String jobjson = getJobs(cookie, null, (String)map.get("name"));
                List<Map<String, Object>> joblist = JsonUtil.fromJsonArray(jobjson);
                if(joblist != null && joblist.size() > 0) {
                    for (Map<String, Object> job : joblist) {
                        if(map.get("last_modified") == null || ((Double)map.get("last_modified")).longValue() <= lasttime){
                            continue;
                        }
                        job.put("cubeId", map.get("uuid"));
                        jobs.add(job);
                    }
                }

                /* 根据segment单条获取执行任务数据，但数据中缺少exec_start_time和exec_end_time
                List<Map<String, Object>> segments = (List<Map<String, Object>>)map.get("segments");
                for(Map<String, Object> segment : segments){
                    String jobjson = getJobs(cookie, (String)segment.get("last_build_job_id"), null);
                    Map<String, Object> job = JsonUtil.fromJson(jobjson);
                    if(job != null) {
                        job.put("cubeId", map.get("uuid"));
                        jobs.add(job);
                    }
                }
                */
                String cuboidjson = getCuboids(cookie, (String)map.get("name"));
                Map<String, Object> cids = JsonUtil.fromJson(cuboidjson);
                if(cids != null) {
                    cids.put("cubeId", map.get("uuid"));
                    cids.put("last_modified", map.get("last_modified"));
                    cuboids.add(cids);
                }
            }
        }
        kylinData.setCubes(cubes);

        //4.处理job
        it = jobs.iterator();
        List<Map<String, Object>> jobOutputs = new ArrayList<>();
        while(it.hasNext()){
            Map<String, Object> map = it.next();
            if(map == null || map.get("last_modified") == null || ((Double)map.get("last_modified")).longValue() <= lasttime){
                it.remove();
            }else{
                /*
                List<Map<String, Object>> steps = (List<Map<String, Object>>)map.get("steps");
                for(Map<String, Object> step : steps){
                    String output = getJobsOutput(cookie, (String)map.get("uuid"), (String)step.get("id"));
                    jobOutputs.add(JsonUtil.fromJson(output));
                }
                */
            }
        }
        kylinData.setJobs(jobs);
        kylinData.setJobOutputs(jobOutputs);

        //5.处理cuboids
        kylinData.setCuboids(cuboids);

        //6. 处理hivetable
        it = tables.iterator();
        while(it.hasNext()){
            Map<String, Object> map = it.next();
            if(map == null || map.get("last_modified") == null || ((Double)map.get("last_modified")).longValue() <= lasttime){
                it.remove();
            }
        }
        kylinData.setTables(tables);

        //7. 处理model
        it = models.iterator();
        while(it.hasNext()){
            Map<String, Object> map = it.next();
            if(map == null || map.get("last_modified") == null || ((Double)map.get("last_modified")).longValue() <= lasttime){
                it.remove();
            }
        }
        kylinData.setModels(models);

        kylinData.setLastModified(lasttime);
        return kylinData;
    }

}
